package cn.antcore.kafka.spring;

import cn.antcore.kafka.annotation.KafkaClient;
import cn.antcore.kafka.builder.KafkaBuilder;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Component;

/**
 * Spring Kafka，建立KafkaClient
 * Created By Hong on 2018/8/21
 **/
@Component
public class SpringKafka implements ApplicationContextAware, InitializingBean {

    private ConfigurableApplicationContext context;

    /**
     * Bean工厂
     */
    private ConfigurableListableBeanFactory beanFactory;

    @Override
    public void afterPropertiesSet() {
        init();
    }

    /**
     * 初始化
     */
    public void init() {
        this.beanFactory = context.getBeanFactory();
        scanKafkaClient();
    }

    /**
     * 获取全部KafkaClient并创建
     */
    private void scanKafkaClient() {
        String[] names = beanFactory.getBeanNamesForAnnotation(KafkaClient.class);
        KafkaBuilder.createClient(beanFactory, names);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = (ConfigurableApplicationContext) applicationContext;
    }
}
